package net.bwie.dwd;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

// 退款原因汇总

public class Dwd_refund_reason {
    public static void main(String[] args) {
        // 1.创建环境
        TableEnvironment tableEnv = getTableEnv();

        // 2.创建输入表
        createInputTable(tableEnv);

        // 3.创建输出表
        createoutputTable(tableEnv);

        // 4.处理数据
        Table resultTable = handle(tableEnv);

        // 5.保存数据
        saveToSink(tableEnv, resultTable);
    }
    private static TableEnvironment getTableEnv() {
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnvironment = TableEnvironment.create(build);

        Configuration configuration = tableEnvironment.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");

        return tableEnvironment;
    }

    private static void createInputTable(TableEnvironment tableEnv) {


        //order_base（订单基础信息表）
        tableEnv.executeSql("CREATE TABLE  ods_order_base (\n" +
                "order_id STRING, \n"+
                "user_id STRING, \n"+
                "shop_id STRING, \n"+
                "product_id STRING, \n"+
                "pay_time TIMESTAMP(3), \n"+
                "promise_ship_time TIMESTAMP(3), \n"+
                "click_ship_time TIMESTAMP(3), \n"+
                "order_amount DOUBLE, \n"+
                "order_status STRING\n"+
                ")WITH(\n"+
                "'connector' = 'kafka',\n" +
                "'topic' = 'ods-order-base-Log',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'properties.group.id' = 'ods-order-base-Log-id01',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")"
        );
//        TableResult tableResult2 = tableEnv.executeSql("SELECT * FROM ods_order_base");
//        tableResult2.print();

        //refund_detail（退款明细信息表）
        tableEnv.executeSql("CREATE TABLE ods_refund_detail (\n" +
                "refund_detail_id STRING, \n"+
                "refund_id STRING, \n"+
                "order_id STRING, \n"+
                "user_id STRING, \n"+
                "apply_time TIMESTAMP(3), \n"+
                "refund_status STRING, \n"+
                "finish_time TIMESTAMP(3), \n"+
                "product_id STRING, \n"+
                "refund_amount DOUBLE, \n"+
                "user_fill_reason STRING, \n"+
                "refund_scene STRING \n"+
                ")WITH(\n"+
                "'connector' = 'kafka',\n" +
                "'topic' = 'ods-refund-detail-log',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'properties.group.id' = 'ods-refund-detail-log-id01',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")"
        );
//        TableResult tableResult3 = tableEnv.executeSql("SELECT * FROM ods_refund_detail");
//        tableResult3.print();

        //consumer_behavior（消费者行为表）
        tableEnv.executeSql("CREATE TABLE ods_consumer_behavior (\n" +
                "behavior_id STRING, \n"+
                "user_id STRING, \n"+
                "behavior_type STRING, \n"+
                "behavior_time TIMESTAMP(3), \n"+
                "behavior_scenario STRING \n"+
                ")WITH(\n"+
                "'connector' = 'kafka',\n" +
                "'topic' = 'ods-consumer-behavior-log',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'properties.group.id' = 'ods-consumer-behavior-log-id01',\n" +
                "'scan.startup.mode' = 'earliest-offset',\n" +
                "'format' = 'json'\n" +
                ")"
        );
//        TableResult tableResult4 = tableEnv.executeSql("SELECT * FROM ods_consumer_behavior");
//        tableResult4.print();

    }

    private static void createoutputTable(TableEnvironment tableEnv) {

        //dwd_retreat_behavior（退后行为汇总表）
        tableEnv.executeSql("CREATE TABLE dwd_refund_reason (\n" +
                //（订单id）
                "order_id STRING ,\n" +
                // （消费者id）
                "user_id STRING ,\n" +
                // (商品ID)
                "product_id STRING ,\n" +
                // (订单总金额)
                "order_amount DOUBLE ,\n" +
                //(订单支付时间)
                "pay_time TIMESTAMP(3) ,\n" +

                // （退款单号）
                "refund_id STRING ,\n" +
                // (实际退款金额)
                "refund_amount DOUBLE ,\n" +

                // （消费者行为id）
                "behavior_id STRING ,\n" +
                // （行为类型）
                "behavior_type STRING, \n" +

                "PRIMARY KEY (order_id) NOT ENFORCED\n"+
                ")WITH(\n" +
                "'connector' = 'upsert-kafka',\n" +
                "'topic' = 'dwd_refund_reason',\n" +
                "'properties.bootstrap.servers' = 'node101:9092',\n" +
                "'key.format' = 'json',\n" +
                "'value.format' = 'json'\n" +
                ")");
    }

    private static Table handle(TableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery("SELECT \n" +
                "ods_order_base.order_id,\n" +
                "ods_order_base.user_id,\n" +
                "ods_order_base.product_id,\n" +
                "ods_order_base.order_amount,\n" +
                "ods_order_base.pay_time,\n" +

                "ods_refund_detail.refund_id,\n" +
                "ods_refund_detail.refund_amount,\n" +

                "ods_consumer_behavior.behavior_id,\n" +
                "ods_consumer_behavior.behavior_type\n" +

                //order_base（订单基础信息表）
                "FROM ods_order_base " +
                //refund_detail（退款明细信息表）
                "left join ods_refund_detail on ods_order_base.order_id = ods_refund_detail.order_id \n"+
                //consumer_behavior（消费者行为表）
                "left join ods_consumer_behavior on ods_consumer_behavior.user_id = ods_order_base.user_id \n"+
                "where ods_refund_detail.apply_time  >=  ods_order_base.pay_time \n"
        );
        return table;
    }

    private static void saveToSink(TableEnvironment tableEnv, Table resultTable) {
        tableEnv.createTemporaryView("resultTable", resultTable);
        tableEnv.executeSql("INSERT INTO dwd_refund_reason SELECT * FROM resultTable");
    }
}
